package com.zt.bigdata.flink.stream

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

/**
  * @ClassName KafkaStream
  * @Description
  * @Author zhangtonghy
  * @Date 2019-07-10 14:26
  * @Copyright: 版权所有 (C) zt zt.
  * @注意 ：本内容仅限于zt内部传阅，禁止外泄以及用于其他的商业目的
  **/
object KafkaStream {
  def main(args: Array[String]): Unit = {
    val see = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "172.30.12.247:9092")
    // only required for Kafka 0.8
    properties.setProperty("zookeeper.connect", "172.30.12.247:2181")
    properties.setProperty("group.id", "test")
    val lines = see
      .addSource(new FlinkKafkaConsumer[String]("flink-1", new SimpleStringSchema(), properties))

    val word = lines.flatMap {
      _
        .toLowerCase
        .split("\\s+")
        .filter(_.nonEmpty)
    }
    val pair = word.map((_, 1))
    val wordCount = pair
      .keyBy(0)
      .timeWindow(Time.seconds(10), Time.seconds(5))
      .sum(1)
    wordCount.print()
    see.execute("kafka streaming")

  }
}
